package com.stone.design.mode.observer;

import com.stone.starter.core.util.thread.AsyncTask;
import com.stone.starter.core.util.thread.CommonExecutor;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;

/**
 * 支持异步执行的事件总线，注意观察者执行顺序无法保证
 *
 * @author Mr_wenpan@163.com 2021/11/30 9:24 下午
 */
public class AsyncEventBus<T> extends EventBus<T> {

    private final ThreadPoolExecutor executor;

    public AsyncEventBus(String identifier, List<? extends Observer<T>> observers, ThreadPoolExecutor executor) {
        super(identifier, observers);
        this.executor = executor;
    }

    @Override
    public void notifyObservers(T target) {
        notifyObservers(target, new Object[]{});
    }

    @Override
    public void notifyObservers(T target, Object... args) {
        if (CollectionUtils.isEmpty(observers)) {
            return;
        }

        // 每一个观察者都创建一个异步任务
        List<AsyncTask<Boolean>> tasks = super.observers.stream().map(o -> new AsyncTask<Boolean>() {
            @Override
            public String taskName() {
                return o.getClass().getSimpleName();
            }

            @Override
            public Boolean doExecute() {
                o.onListener(target, args);
                return Boolean.TRUE;
            }
        }).collect(Collectors.toList());

        // 异步通知每一个观察者
        CommonExecutor.batchExecuteAsync(tasks, executor, identifier);
    }
}